//
// message.cpp
// libprotea
// Tools Library for the Protea Project / Message handling between processes
// Copyright (C) 2003, 2004, 2005 eXegeSys, Inc.
// Copyright (C) 2008 Bruce A. James
//
// This program is part of the Protea Project.
//
// The Protea Project is free software; you can redistribute it and/or modify it 
// under the terms of the GNU General Public License as published by the 
// Free Software Foundation, either version 3 of the License, or (at your 
// option) any later version.
//
// The Protea Project is distributed in the hope that it will be useful, but 
// WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY 
// or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License 
// for more details.
// 
// You should have received a copy of the GNU General Public License along 
// with this program.  If not, see <http://www.gnu.org/licenses/>.
// 
// Please send correspondence to:
// 
// theproteaproject@gmail.com
//

#include "../defines.h"
#include "../OSDep/transport.h"
#include "AppInterface.h"
#include "message.h"
#include "RLE.h"


#define NO_COMPRESSION

/*
	\file message.cpp
	\brief Message Class

	Includes definitions of the XCToolsMessage class which is the basis of all messages
	sent between any tools processes
*/




//
// Helper functions
//

/*
	\brief Helper function to align size on dword boundarys
	\param size Size to be aligned

	Result is the aligned size

	This MUST be done to prevent bus errors on RISC platforms
*/
long Align( long size ) {
	int packing = 2;
	long temp = size;
	temp += (1 << packing) - 1;
	temp >>= packing;
	temp <<= packing;

	return temp;
};





////////////////////////////////////////////////////////////////
//
//	XCToolsMessage Class
//
/////////////////////////////////////////////////////////////////
XCToolsMessage::XCToolsMessage( XCAppInterface* owner ) {
	msg = NULL;
	compbuf = NULL;

	myowner = owner;
};

XCToolsMessage::~XCToolsMessage() {
	DeleteMsg();
};




XCToolsMessage* XCToolsMessage::Copy() {
	XCToolsMessage* result = new XCToolsMessage( myowner );

	unsigned long size = GetDataFullSize();
	result->msg = (base_message*)new char[ size ];
	memcpy( result->msg, msg, size );
	return result;
};





/**
	\param size Message size to be allocated

	Note: The size parameter must NOT include the constant BASE_MESSAGE_SIZE
*/
void XCToolsMessage::CreateMsg( long size ) {

	INFUNC( XCToolsMessage::CreateMsg, myowner )

//	long aligned = Align( size ) + BASE_MESSAGE_SIZE;
	long aligned = size + BASE_MESSAGE_SIZE;

	DeleteMsg();

	msg = (base_message*) new char[ aligned ];
	memset( msg, 0, aligned );
	msg->orig_size = size;
	msg->data_size = size;
	msg->type = (message_type)(-1);	// invalid type
	msg->sdflag = src_dest_clear;
	msg->miscflag = misc_clear;

	OUTFUNC()
};


void XCToolsMessage::CreateMsg( base_message* header ) {
	INFUNC( XCToolsMessage::CreateMsg, myowner )

	DeleteMsg();

	myowner->ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "Align: header = %x", header );

//	long size = Align( header->orig_size ) + BASE_MESSAGE_SIZE;
	long size = header->orig_size + BASE_MESSAGE_SIZE;

	myowner->ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "CreateMsg: size = %ld", size );

	msg = (base_message*) new char[ size ];

	myowner->ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "Msg: %x", msg );

	memcpy( msg, header, BASE_MESSAGE_SIZE );

	myowner->ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "Done" );

	// Check to see if we need to allocate a compression buffer
	if (header->orig_size != header->data_size)
		compbuf = new char[ header->data_size ];
//		compbuf = new char[ Align( header->data_size ) ];

	OUTFUNC()
};




void XCToolsMessage::DeleteMsg() {
	INFUNC( XCToolsMessage::DeleteMsg, myowner )

	// Recover memory from the actual message
	if (msg) {
		delete [] msg;
		msg = NULL;
	};

	// Recover memory from the comp/decomp buffer
	if (compbuf) {
		delete [] compbuf;
		compbuf = NULL;
	};

	OUTFUNC()
};




void XCToolsMessage::ChangeMsgSize( long size )
{
	base_message* retval = NULL;
	
	INFUNC( ChangeMsgSize, myowner )

	// Calculate the size with route count
//	long oldsz = Align(msg->data_size);
//	long newsz = Align(oldsz + size);
	long oldsz = msg->data_size;
	long newsz = oldsz + size;

	myowner->ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "Change Size old message size %ld", oldsz );
	myowner->ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "Change Size new message size %ld", newsz );

	retval = (base_message*) new char[newsz];

	if (size < 0 )
		memcpy( retval, msg, newsz );
	else
		memcpy( retval, msg, oldsz );

	DeleteMsg();
	msg = retval;

	OUTFUNC()
};




void* XCToolsMessage::GetDataPtr() const {
	return (char*)msg + BASE_MESSAGE_SIZE;
};

base_message* XCToolsMessage::GetHeaderPtr() const {
	return msg;
};


void XCToolsMessage::SetHeader( message_type type, route_loc src, route_loc dest, bool broadcast, bool query )
{
	INFUNC( XCToolsMessage::SetHeader, myowner )

	// Set message type
	msg->type = type;

	// Set source and destination
	msg->sdflag &= src_dest_clear;
	msg->sdflag |= src;
	msg->sdflag |= dest;

	// Set misc flags
	msg->miscflag &= misc_clear_other;
	if ( broadcast )
		msg->miscflag |= misc_broadcast;
	if ( query )
		msg->miscflag |= misc_query;

	OUTFUNC()
};




void XCToolsMessage::SetQueryResponse( bool query ) {
	if (query)
		msg->miscflag |= misc_query;
	else
		msg->miscflag &= !misc_query;
};




void XCToolsMessage::SetHeader( message_type type )
{
	INFUNC( XCToolsMessage::SetHeader, myowner )

	// Set message type
	msg->type = type;

	OUTFUNC()
};


void XCToolsMessage::AddRouteInfo( const XCMsgPort* src, const XCMsgPort* dest )
{
	INFUNC( XCToolsMessage::AddRouteInfo, myowner )

	// increment route count
	unsigned char temp = msg->miscflag & misc_clear_other;
	route_info* myroute = &msg->routes[ temp++ ];

	// Check maximum route size
	assert( temp <= MAX_ROUTES );

	msg->miscflag &= misc_clear_count;
	msg->miscflag |= temp;
	myowner->ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "Current route count : %d", temp );

	if (src)
		myroute->src = (long)src->transport->GetInHandle();
	else
		myroute->src = 0;

	if (dest)
		myroute->dest = (long)dest->transport->GetOutHandle();
	else
		myroute->dest = 0;

	myowner->ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "Add route %ld to %ld", myroute->src, myroute->dest );
	myowner->ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "Done writing route" );

	OUTFUNC()
};





void XCToolsMessage::ClearRoutes()
{
	INFUNC( XCToolsMessage::ClearRoutes, myowner );

	// Clear route count
	msg->miscflag &= misc_clear_count;

	OUTFUNC();
};




void XCToolsMessage::RemoveRouteInfo()
{
	INFUNC( XCToolsMessage::RemoveRouteInfo, myowner )

	// decrement route count
	unsigned char temp = msg->miscflag & misc_clear_other;
	temp--;

	// Check minimum route size
	assert( temp <= MAX_ROUTES );

	// Update flags
	msg->miscflag &= misc_clear_count;
	msg->miscflag |= temp;

	OUTFUNC()
};





void XCToolsMessage::CopyRouteInfo( const XCToolsMessage& src )
{
	INFUNC( XCToolsMessage::CopyRouteInfo, myowner )

	// check for route count = 0
	unsigned char temp = msg->miscflag & misc_clear_other;
	if (temp != 0) {
		myowner->ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "CopyRouteInfo : Destination Route Info exists" );
		RAISETOOLSERROR( -1 );
	}

	// Copy route count
	temp = src.msg->miscflag & misc_clear_other;
	msg->miscflag |= temp;

	// Copy routing table
	memcpy( msg->routes, src.msg->routes, MAX_ROUTES * sizeof( route_info ) );

	OUTFUNC()
};


void XCToolsMessage::SetReply( const XCToolsMessage& src )
{
	INFUNC( XCToolsMessage::SetReply, myowner )

	CopyRouteInfo( src );

	msg->type = src.msg->type;
	msg->sdflag = src_dest_clear;
	
	// swap src flags
	if (src.msg->sdflag & src_gcs)
	{
		msg->sdflag |= dest_gcs;
		myowner->ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "SetReply src_gcs => dest_gcs" );
	}

	if (src.msg->sdflag & src_app)
	{
		msg->sdflag |= dest_app;
		myowner->ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "SetReply src_app => dest_app" );
	}
	if (src.msg->sdflag & src_int)
	{
		msg->sdflag |= dest_int;
		myowner->ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "SetReply src_int => dest_int" );
	}

	
	// swap dest flags
	if (src.msg->sdflag & dest_gcs)
	{
		msg->sdflag |= src_gcs;
		myowner->ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "SetReply dest_gcs => src_gcs" );
	}
	if (src.msg->sdflag & dest_app)
	{
		msg->sdflag |= src_app;
		myowner->ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "SetReply dest_app => src_app" );
	}
	if (src.msg->sdflag & dest_int)
	{
		msg->sdflag |= src_int;
		myowner->ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "SetReply dest_int => src_int" );
	}

	// set misc flag
	msg->miscflag = src.msg->miscflag % misc_query;

	OUTFUNC()
};




route_info* XCToolsMessage::PeekLastRoute() const {
	unsigned char last = msg->miscflag & misc_clear_other;
	return &msg->routes[ last-1 ];
};



long XCToolsMessage::GetDataFullSize() const {
//	return Align(msg->data_size) + BASE_MESSAGE_SIZE;
	return msg->data_size + BASE_MESSAGE_SIZE;
};





void XCToolsMessage::Compress() {
	INFUNC( XCToolsMessage::Compress, myowner )

	msg->data_size = msg->orig_size;

	assert( msg->data_size > 0 );

	compbuf = new char[ msg->data_size ];

#ifdef NO_COMPRESSION
	// Fake compression
	memcpy( compbuf, GetDataPtr(), msg->orig_size );
#else
	// Attempt compression of the data
	msg->data_size = RLECompress(
		(unsigned char*)GetDataPtr(), msg->orig_size,
		(unsigned char*)compbuf, msg->data_size
	);

	// If compression failed, cleanup
	if (msg->data_size == 0) {
		msg->data_size = msg->orig_size;
		delete [] compbuf;
		compbuf = NULL;
	};
#endif

	OUTFUNC();
};


void XCToolsMessage::Decompress() {
	INFUNC( XCToolsMessage::Decompress, myowner )

	if (!UsesCompression())
		EXITFUNC();

#ifdef NO_COMPRESSION
	memcpy( GetDataPtr(), compbuf, msg->orig_size );
#else
	// Run the decompression routine
	msg->data_size = RLEUncompress(
		(unsigned char*)compbuf, msg->data_size,
		(unsigned char*)GetDataPtr(), msg->orig_size
	);

//	cout << msg->data_size << " <===> " << msg->orig_size << endl;

	assert( msg->data_size == msg->orig_size );
#endif

	OUTFUNC();
};



unsigned char XCToolsMessage::GetMsgChar( unsigned long offset ) {
	if ( offset < BASE_MESSAGE_SIZE || !UsesCompression() )
		return ((unsigned char*)msg)[ offset ];
	else
		return compbuf[ offset - BASE_MESSAGE_SIZE ];
};



void XCToolsMessage::Dump()
{
	unsigned long total_size = GetDataFullSize();
	unsigned long startpos = 0;
	char from[20];
	char temps[ 80 ];
	long temp;
	bool is_transaction = false;

	myowner->ToolsTrace( TRC_TRAFFIC, TLV_LOW, "-----------------------------" );
	myowner->ToolsTrace( TRC_TRAFFIC, TLV_LOW, "         MESSAGE DUMP" );
	myowner->ToolsTrace( TRC_TRAFFIC, TLV_LOW, "-----------------------------" );
	myowner->ToolsTrace( TRC_TRAFFIC, TLV_MEDIUM, "Total size: %ld", total_size );
	myowner->ToolsTrace( TRC_TRAFFIC, TLV_LOW, "Data size : %ld", msg->data_size );
	myowner->ToolsTrace( TRC_TRAFFIC, TLV_MEDIUM, "Orig size : %ld", msg->orig_size );
	temp = msg->type;
	switch( msg->type ) {
		case msg_init:
			sprintf( temps, "INIT" );
			break;
		case msg_user:
			sprintf( temps, "USER" );
			break;
		case msg_admin:
			sprintf( temps, "ADMIN" );
			break;
		case msg_shutdown:
			sprintf( temps, "SHUTDOWN" );
			break;
		case msg_status:
			sprintf( temps, "STATUS" );
			break;
		case msg_sql_query:
		case msg_trans_response:
		case msg_trans: {
			switch( msg->type ) {
				case msg_trans:
					strcpy( from, "QUERY" );
					break;

				case msg_sql_query:
					strcpy( from, "SQL QUERY" );
					break;

				default:
					strcpy( from, "RESPONSE" );
					break;
			};

			trans_message* trans = (trans_message*)GetDataPtr();
			switch( trans->data_type ) {
				case TRANS_HEADER_INFO:
					sprintf( temps, "TRANS %s (HEADER INFO)", from );
					break;
				case TRANS_HEADER_TYPE:
					sprintf( temps, "TRANS %s (HEADER)", from );
					break;
				case TRANS_DETAIL_TYPE:
					sprintf( temps, "TRANS %s (DETAIL)", from );
					break;
				case TRANS_MESSAGE_TYPE:
					sprintf( temps, "TRANS %s (MESSAGE)", from );
					break;
				case QUERY_HEADER_INFO:
					sprintf( temps, "%s (HEADER)", from );
					break;
				case QUERY_FIELD_INFO:
					sprintf( temps, "%s (MESSAGE)", from );
					break;
			};

			// Start output of data at actual data
			startpos += offsetof( trans_message, data );

			// Signal that this is a transaction
			is_transaction = true;
			break;
		};
		case msg_dispatch: {
			sprintf( temps, "DISPATCH: " );

			disp_message* dmsg = (disp_message*)GetDataPtr();
			switch( dmsg->opcode ) {
				case disp_start_app:
					strcat( temps, "START APP" );
					break;
		
				/// shutdown all
				case disp_shutdown:
					strcat( temps, "SHUTDOWN ALL" );
					break;
		
				/// dispatcher to contact ui on port specified
				case disp_contact_ui:
					strcat( temps, "CONTACT UI" );
					break;
		
				/// application to return information to GCS
				case disp_query_info:
					strcat( temps, "QUERY INFO" );
					break;
		
				/// query reply information
				case disp_query_reply:
					strcat( temps, "QUERY REPLY" );
					break;
		
				/// all processes to return information to GCS
				case disp_pushdown_query:
					strcat( temps, "PUSHDOWN QUERY" );
					break;

				/// ???
				case disp_pushdown_start:
					strcat( temps, "PUSHDOWN START" );
					break;
		
				case disp_shutdown_confirm:
					strcat( temps, "SHUTDOWN CONFIRM" );
					break;

				case disp_hello:
					strcat( temps, "HELLO" );
					break;

				default:
					sprintf( &temps[ strlen( temps ) ], "UNKNOWN: %ld", dmsg->opcode );
					break;
			};
			break;
		};
		case msg_options:
			sprintf( temps, "OPTIONS" );
			break;
		default:
			sprintf( temps, "UNKNOWN" );
			break;
	};

	// Now, output the type
	myowner->ToolsTrace( TRC_TRAFFIC, TLV_LOW, "type      : %s  (%d)", temps, msg->type );

	// Output transaction specifics...
	if (is_transaction) {
		trans_message* trans = (trans_message*)GetDataPtr();
		myowner->ToolsTrace( TRC_TRAFFIC, TLV_MEDIUM, "\tID        : %d", trans->transid );
		myowner->ToolsTrace( TRC_TRAFFIC, TLV_MEDIUM, "\tAppl      : %d", trans->appnbr );
		myowner->ToolsTrace( TRC_TRAFFIC, TLV_MEDIUM, "\tOrder     : %d", trans->order );
		myowner->ToolsTrace( TRC_TRAFFIC, TLV_MEDIUM, "\tFormat    : %d", trans->format );

		// If header record, output even more
		if (trans->data_type == TRANS_HEADER_INFO) {
			trans_info_message* info = (trans_info_message*)trans->data;
			myowner->ToolsTrace( TRC_TRAFFIC, TLV_MEDIUM, "\tTxn number: %d", info->trans_number );
			myowner->ToolsTrace( TRC_TRAFFIC, TLV_MEDIUM, "\tSuccess   : %d", info->success );
			myowner->ToolsTrace( TRC_TRAFFIC, TLV_MEDIUM, "\tPhase     : %d", info->phase );
		};
	};


	// Return flags
	temp = msg->sdflag;
	if (temp & src_gcs)
		strcpy( temps, "GCS" );
	else if (temp & src_app)
		strcpy( temps, "APPLICATION" );
	else if (temp & src_int)
		strcpy( temps, "INTERFACE" );
	else
		strcpy( temps, "UNKNOWN" );

	strcat( temps, " --> " );

	if ((temp & dest_all) == dest_all)
		strcat( temps, "ALL" );
	else if (temp & dest_gcs)
		strcat( temps, "GCS" );
	else if (temp & dest_app)
		strcat( temps, "APPLICATION" );
	else if (temp & dest_int)
		strcat( temps, "INTERFACE" );
	else
		strcat( temps, "UNKNOWN" );

	myowner->ToolsTrace( TRC_TRAFFIC, TLV_LOW, "sdflag    : %s  (%d)", temps, temp );


	// Output misc flags
	*temps = 0;
	if (msg->miscflag & misc_broadcast)
		strcpy( temps, "Broadcast " );

	if (msg->miscflag & misc_query)
		strcat( temps, "Query" );
	else
		strcat( temps, "Reply" );
	myowner->ToolsTrace( TRC_TRAFFIC, TLV_LOW, "miscflag  : %s  (%d)", temps, msg->miscflag );

	myowner->ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "Hex dump  : " );

	unsigned long output_size = total_size - startpos;
	for ( unsigned long x = startpos; x < output_size; x += 8 )
	{
		unsigned long w;

		char temp[ 80 ];
		char* ptemp = temp;

		ptemp += sprintf( ptemp, "%3.3ld : ", x );
		for ( w = 0; w < 8; w++ ) {
			if ( w + x < output_size )
				ptemp += sprintf( ptemp, "%2.2X ", GetMsgChar( w + x ) );
			else
				ptemp += sprintf( ptemp, "   " );
		};

		ptemp += sprintf( ptemp, " - " );
		for ( w = 0; w < 8; w++ ) {
			if ( w + x < output_size ) {
				unsigned char ctemp = GetMsgChar( w+x );
				if ( ctemp < 32 )
					ptemp += sprintf( ptemp, "~" );
				else
					ptemp += sprintf( ptemp, "%c", ctemp );
			};
		};
		myowner->ToolsTrace( TRC_TRAFFIC, TLV_HIGH, temp );
	};
	
	myowner->ToolsTrace( TRC_TRAFFIC, TLV_MEDIUM, "Routing information :" );


	// Output route information
	unsigned char count = msg->miscflag & misc_clear_other;
	route_info* route = msg->routes;
	while (count) {
		myowner->ToolsTrace( TRC_TRAFFIC, TLV_MEDIUM, "-- From : %ld  -- To : %ld", route->src, route->dest );
		route++;
		count--;
	};
	myowner->ToolsTrace( TRC_TRAFFIC, TLV_LOW, "-----------------------------" );
};



////////////////////////////////////////////////////////////////
//
//	XCMsgPort Class
//
///////////////////////////////////////////////////////////////
XCMsgPort::XCMsgPort( process_info* process, XCAppInterface* owner ) {
	INFUNC( XCMsgPort::XCMsgPort, owner );

	new_connection = NULL;
	requests = 0;
	killme = false;

	awaiting_header = true;

	buffer = NULL;
	message = new XCToolsMessage( owner );
	myowner = owner;
	if (process)
	{
		myprocess = process;
		process->port = this;
	}
	else
	{
		myprocess = NULL;
	}
	next = NULL;
	received = 0;
	expected = 0;
	extradata = NULL;

	OUTFUNC()
};



XCMsgPort::~XCMsgPort() {
	INFUNC( XCMsgPort::~XCMsgPort, myowner );

	if (new_connection)
		delete new_connection;

	if (transport) {
		if (myowner) 
			myowner->ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "Deleting msg port: %ld", transport->GetInHandle() );
		transport->CancelIO();
	};

	ClearBuffer();

	if (transport)
		delete transport;

	delete message;

	OUTFUNC()
};


void XCMsgPort::AllocateBuffer( int size ) {
	// Clear any existing buffers that we own
	ClearBuffer();

	// Allocate a new buffer
	buffer = new char[ size ];

	// Set out expected/received flags
	expected = size;
	received = 0;
};

void XCMsgPort::ClearBuffer() {
	if (buffer) {
		delete [] buffer;
		buffer = NULL;
	};
};


void XCMsgPort::Release() {
	INFUNC( XCMsgPort::Release, myowner );

	if (transport && !killme) {
		if (myowner)
			myowner->ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "Release message port: %ld", (long)transport->GetInHandle() );

		if (requests) {
			transport->CancelIO();
			transport->Close();
			killme = true;
		} else
			delete this;
	} else
		delete this;

	OUTFUNC()
};



void XCMsgPort::SetTransport( XCTransport* new_transport ) {
	transport = new_transport;
	transport->SetMsgPort( this );
};
